package com.xiao.dynamic_rule.service;

import com.xiao.dynamic_rule.pojo.BufferResult;
import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.RuleParam;
import com.xiao.dynamic_rule.pojo.RuleParamAtomic;
import com.xiao.dynamic_rule.utils.RuleCalcUtil;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.state.ListState;

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

/**
 * @description: 分路查询路由(加入缓存)
 * @author: 啸
 * @date: 2022/8/23 10:11
 */
public class QueryRouterV4 {

    UserProfileQueryServiceHbaseImpl userProfileQueryServiceHbase;
    UserActionCountQueryServiceStateImpl userActionCountQueryServiceState;
    UserActionCountQueryServiceClickHouseImpl userActionCountQueryServiceClickHouse;
    UserActionSequenceQueryServiceStateImpl userActionSequenceQueryServiceState;
    UserActionSequenceQueryServiceClickhouseImpl userActionSequenceQueryServiceClickhouse;
    BufferManager bufferManager;


    // 在构造方法中初始化连接
    public QueryRouterV4() throws IOException, SQLException, ClassNotFoundException {
        userProfileQueryServiceHbase = new UserProfileQueryServiceHbaseImpl();
        userActionCountQueryServiceState = new UserActionCountQueryServiceStateImpl();
        userActionCountQueryServiceClickHouse = new UserActionCountQueryServiceClickHouseImpl();
        userActionSequenceQueryServiceState = new UserActionSequenceQueryServiceStateImpl();
        userActionSequenceQueryServiceClickhouse = new UserActionSequenceQueryServiceClickhouseImpl();
        bufferManager = new BufferManager();
    }

    /**
     * 画像查询路由
     *
     * @param ruleParam 规则条件
     * @param logBean   flink stream
     * @return 是否满足
     */
    public boolean userProfileQuery(RuleParam ruleParam, LogBean logBean) {
        //画像条件
        boolean profileMatch = userProfileQueryServiceHbase.judgeQueryProfileCondition(logBean.getDeviceId(), ruleParam);
        return profileMatch;
    }

    /**
     * 行为次数查询路由
     * TODO 将规则条件分开查询，最后没有累加？？？
     */
    public boolean countConditionQuery(RuleParam ruleParam, LogBean logBean, ListState<LogBean> eventState) throws Exception {
        // 拿到条件中的次数条件
        List<RuleParamAtomic> userActionCountParams = ruleParam.getUserActionCountParams();

        /**
         * 引入缓存处理
         */
        updateRuleParamByBufferResult(logBean, userActionCountParams);

        // 获取分界点（当前时间向下取整，再向前减1小时）
        long splitPoint = DateUtils.addHours(DateUtils.ceiling(new Date(), Calendar.HOUR), -2).getTime();

        // 为了减小clickhouse的压力，将条件根据分界点分为三段，
        // startTime < 分界点就去clickhouse中查，startTime > 分界点就在state中查
        // 时间跨度包含分界点，则先去state查，如果不满足则在去clickhouse中查，然后state结果和clickhouse结果整合来判断是否满足条件
        ArrayList<RuleParamAtomic> nearRangeParams = new ArrayList<>();
        ArrayList<RuleParamAtomic> farRangeParams = new ArrayList<>();
        ArrayList<RuleParamAtomic> crossRangeParams = new ArrayList<>();


        // 根据时间来分为三组
        for (RuleParamAtomic userActionCountParam : userActionCountParams) {
            if (userActionCountParam.getRangeEnd() < splitPoint) {
                // 放到nearRangeParams中
                nearRangeParams.add(userActionCountParam);
            } else if (userActionCountParam.getRangeStart() >= splitPoint) {
                // 放到farRangeParams中
                farRangeParams.add(userActionCountParam);
            } else {
                // 放到crossRangParams中
                crossRangeParams.add(userActionCountParam);
            }
        }

        /**
         * 先查近期查询，如果在state中不满足，则可以不必去clickhouse中查，减少查询，提高速度
         */
        if (nearRangeParams.size() > 0) {
            //将规则总参数对象中的用户次数类规则覆盖为 近期条件组
            ruleParam.setUserActionCountParams(nearRangeParams);
            // 在state中查询
            boolean nearCountMatch = userActionCountQueryServiceState.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            // 将查询结果放入缓存
            for (RuleParamAtomic userActionCountParam : userActionCountParams) {
                String bufferKey = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionCountParam);
                bufferManager.putBufferData(bufferKey, userActionCountParam.getRealCunts(), userActionCountParam.getRangeStart(), userActionCountParam.getRangeEnd());
            }

            return nearCountMatch;
        }

        /**
         * 跨界查询
         */
        for (RuleParamAtomic crossRangeParam : crossRangeParams) {
            //首先将原来规则的 startTime和endTime保存下来
            long originRangeStart = crossRangeParam.getRangeStart();
            long originRangeEnd = crossRangeParam.getRangeEnd();

            //获取bufferKey
            String bufferKey = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), crossRangeParam);

            //将规则中的startTime设置成分界点，查state
            crossRangeParam.setRangeStart(splitPoint);
            boolean crossCountStateMatch = userActionCountQueryServiceState.queryActionCount(logBean.getDeviceId(), eventState, crossRangeParam);
            if (crossCountStateMatch) {
                //将查询到的结果放入缓存中
                bufferManager.putBufferData(bufferKey, crossRangeParam.getRealCunts(), crossRangeParam.getRangeStart(), crossRangeParam.getRangeEnd());
                continue;
            }

            //如果上面不满足，则将rangeStart换为原来的开始点，rangeEnd换为分界点，去clickhouse中查询
            crossRangeParam.setRangeStart(originRangeStart);
            crossRangeParam.setRangeEnd(splitPoint);
            boolean crossCountClickhouseMatch = userActionCountQueryServiceClickHouse.queryActionCount(logBean.getDeviceId(), eventState, crossRangeParam);

            //将查询结果放入缓存中，因为到这里已经将完整的条件都进行查询了，所以结果的时间范围就是最初的时间范围
            bufferManager.putBufferData(bufferKey, crossRangeParam.getRealCunts(), originRangeStart, originRangeEnd);

            return crossCountClickhouseMatch;
        }


        /**
         * 远期查询
         */
        if (farRangeParams.size() > 0) {
            //覆盖参数
            ruleParam.setUserActionCountParams(farRangeParams);
            //在clickhouse中查询
            boolean farCountMatch = userActionCountQueryServiceClickHouse.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            // 将远期查询结果放入缓存中
            for (RuleParamAtomic farRangeParam : farRangeParams) {
                String bufferKey = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), farRangeParam);
                bufferManager.putBufferData(bufferKey, farRangeParam.getRealCunts(), farRangeParam.getRangeStart(), farRangeParam.getRangeEnd());
            }
            return farCountMatch;
        }

        return true;
    }


    /**
     * 行为序列查询路由
     */
    public boolean sequenceConditionQuery(RuleParam ruleParam, LogBean logBean, ListState<LogBean> eventState) throws Exception {

        //取出规则中的行为序列条件
        List<RuleParamAtomic> userActionSequenceParams = ruleParam.getUserActionSequenceParams();

        //取出规则中的开始时间，结束时间，需要满足的步骤数
        long rangeStart = userActionSequenceParams.get(0).getRangeStart();
        long rangeEnd = userActionSequenceParams.get(0).getRangeEnd();
        int totalSteps = userActionSequenceParams.size();

        /**
         * 引入缓存处理
         *      根据查询后的结果，如果全部满足，则从规则中删除此条规则，并且将结果写回到结果对象中
         *                     如果部分满足，则将窗口的开始时间设置为缓存的结束时间，然后将缓存中的最大步骤数写道规则对象中
         */
        //将传入的条件拼接成redis中那样的key
        String bufferKey = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);

        //查询缓存
        BufferResult bufferResult = bufferManager.getBufferData(bufferKey, rangeStart, rangeEnd, totalSteps);

        switch (bufferResult.getBufferAvailableLevel()) {

            //如果部分有效：rangeStart相同，则将规则中的开始时间更新为缓存的结束时间
            case PARTIAL_AVL:
                //更新条件窗口的开始时间
                userActionSequenceParams.get(0).setRangeStart(bufferResult.getBufferRangeEnd());
                //将缓存中的最大步骤数值放到ruleParam中的maxStep中
                ruleParam.setUserActionSequenceQueriedMaxStep(bufferResult.getValue());
                // 截短序列
                List<RuleParamAtomic> newSequenceList = userActionSequenceParams.subList(bufferResult.getValue(), totalSteps);
                ruleParam.setUserActionSequenceParams(newSequenceList);
                break;

            case UN_AVL:
                break;

            //如果全部有效，则直接返回true
            case WHOLE_AVL:
                return true;
        }


        //序列条件中有条件才计算
        //将截短后的序列条件更新到序列条件中
        userActionSequenceParams = ruleParam.getUserActionSequenceParams();
        if (userActionSequenceParams != null && userActionSequenceParams.size() > 0) {
            //取到分界点时间（和次数类计算应该有时间差，所以单独创建）
            long splitPoint = DateUtils.addHours(DateUtils.ceiling(new Date(), Calendar.HOUR), -2).getTime();

            //取出缓存中已经查到的最大步骤数，来和后面计算的结果累加，得到正确结果
            int bufferUserActionSequenceQueriedMaxStep = ruleParam.getUserActionSequenceQueriedMaxStep();
            //取出序列规则中的总步骤数
            int totalSize = ruleParam.getUserActionSequenceParams().size();

            if (userActionSequenceParams.get(0).getRangeStart() > splitPoint) {
                //全部在state中查
                boolean b = userActionSequenceQueryServiceState.queryActionSequence(logBean.getDeviceId(), eventState, ruleParam);
                String bufferKey1 = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);
                bufferManager.putBufferData(bufferKey1, ruleParam.getUserActionSequenceQueriedMaxStep(), rangeStart, rangeEnd);
                return b;
            } else if (userActionSequenceParams.get(0).getRangeEnd() < splitPoint) {
                //state中一部分，clickhouse中一部分
                //将条件开始时间换为分界点
                modifyTimeRange(userActionSequenceParams, splitPoint, userActionSequenceParams.get(0).getRangeEnd());
                //先查state，如果state中已经满足了条件，则就不必去clickhouse中查了
                // bug修复：因为加入了缓存的逻辑，在缓存部分就已经将getUserActionSequenceQueriedMaxStep设了一个值，并且将序列条件截短了。
                // 当这个查完之后，会去和截短后的序列条件的size比较，因为缓存就已经赋值了maxStep，所以state中没有满足条件，也可能返回true（可能缓存中完成的maxStep >= 截短厚的条件list.size()）
                userActionSequenceQueryServiceState.queryActionSequence("", eventState, ruleParam);
                if (ruleParam.getUserActionSequenceQueriedMaxStep() >= totalSize) {
                    //在state中满足剩下的条件时，将查询结果写入缓存
                    String bufferKey1 = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);
                    bufferManager.putBufferData(bufferKey1, ruleParam.getUserActionSequenceQueriedMaxStep(), userActionSequenceParams.get(0).getRangeStart(), userActionSequenceParams.get(0).getRangeEnd());
                    return true;
                } else {
                    //如果state不完全满足条件，则将maxStep恢复为原来的值
                    ruleParam.setUserActionSequenceQueriedMaxStep(bufferUserActionSequenceQueriedMaxStep);
                }

                //将条件结束时间换为分界点
                modifyTimeRange(userActionSequenceParams, userActionSequenceParams.get(0).getRangeStart(), splitPoint);
                //如果在state中没有查到，则去clickhouse中查，clickhouse满足也行
                boolean b1 = userActionSequenceQueryServiceClickhouse.queryActionSequence(logBean.getDeviceId(), null, ruleParam);
                if (b1) {
                    //如果完全满足则将查询结果放入缓存中
                    String bufferKey1 = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);
                    bufferManager.putBufferData(bufferKey1, ruleParam.getUserActionSequenceQueriedMaxStep(),userActionSequenceParams.get(0).getRangeStart(), splitPoint);
                    return true;
                }

                //如果clickhouse也不能全部满足，则将条件中clickhouse没查到的条件去state查，然后组合
                //将条件的开始时间换为分界点
                modifyTimeRange(userActionSequenceParams, splitPoint, rangeEnd);

                //现在将clickhouse已经查到满足的条件去除，只在state中查询还未满足的条件
                ruleParam.setUserActionSequenceParams(userActionSequenceParams.subList(ruleParam.getUserActionSequenceQueriedMaxStep(), userActionSequenceParams.size()));

                //查询截断后的条件
                userActionSequenceQueryServiceState.queryActionSequence("", eventState, ruleParam);

                //返回结果
                if (ruleParam.getUserActionSequenceQueriedMaxStep() >= totalSize) {
                    String bufferKey1 = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);
                    bufferManager.putBufferData(bufferKey1, ruleParam.getUserActionSequenceQueriedMaxStep(), rangeStart,rangeEnd);
                    return true;
                } else {
                    ruleParam.setUserActionSequenceParams(userActionSequenceParams);
                    return false;
                }
            }
        } else {
            //传入的条件rangeEnd < splitPoint ，直接在clickhouse中查
            boolean b = userActionCountQueryServiceClickHouse.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            //不管是否完全满足，将查询结果放入缓存
            String bufferKey1 = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionSequenceParams);
            bufferManager.putBufferData(bufferKey1, ruleParam.getUserActionSequenceQueriedMaxStep(), rangeStart,rangeEnd);
        }
        return false;
    }

    /**
     * 更新条件的时间窗口的工具方法
     *
     * @param userActionSequenceParams 行为序列条件
     * @param newStart                 条件中新的开始时间
     * @param newEnd                   条件中新的结束时间
     */
    public void modifyTimeRange(List<RuleParamAtomic> userActionSequenceParams, long newStart, long newEnd) {
        for (RuleParamAtomic userActionSequenceParam : userActionSequenceParams) {
            userActionSequenceParam.setRangeStart(newStart);
            userActionSequenceParam.setRangeEnd(newEnd);
        }
    }

    /**
     * 查询缓存，并根据缓存结果更新规则
     * 1、缩短规则大小
     * 2、剔除条件
     *
     * @param logBean               待处理事件
     * @param userActionCountParams count类条件list
     */
    private void updateRuleParamByBufferResult(LogBean logBean, List<RuleParamAtomic> userActionCountParams) {
        for (int i = 0; i < userActionCountParams.size(); i++) {
            //将传入的条件拼接成redis中那样的key
            String bufferKey = RuleCalcUtil.getBufferKey(logBean.getDeviceId(), userActionCountParams.get(i));

            //查询缓存
            BufferResult bufferResult = bufferManager.getBufferData(bufferKey, userActionCountParams.get(i));

            switch (bufferResult.getBufferAvailableLevel()) {
                //如果全部有效，则剔除该条件
                case WHOLE_AVL:
                    //面试问题点：list中无法直接remove元素，如果要删除的两个元素相邻，则第二个元素将无法删除，因为删除后，再次进入循环前
                    //list会将后面的元素往前移动一位，所以本来下标为3的元素的下标变为2，上回循环就是2，下回循环3，所以就像跳过了这个元素，
                    //所以在判断中加一个i--就可以避免
                    userActionCountParams.remove(i);
                    i--;
                    break;

                //如果部分有效：rangeStart相同，则将规则中的开始时间更新为缓存的结束时间
                case PARTIAL_AVL:
                    //更新条件窗口的开始时间
                    userActionCountParams.get(i).setRangeStart(bufferResult.getBufferRangeEnd());
                    //将缓存中的count值放到ruleParam中的realCnt中
                    userActionCountParams.get(i).setRealCunts(bufferResult.getValue());
                    break;

                case UN_AVL:
            }

        }
    }

}
